package com.dylan.magic.rocketmq.config.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * @author : ankang
 * @date : 2020/1/21 20:32
 * @Description :
 */
@Slf4j
public abstract class BaseMessageConsumer implements InitializingBean {

    @Override
    public void afterPropertiesSet() throws Exception {
        // 注册消费
        subscribe();
    }

    /**
     * 注册消费消息主题
     */
    private void subscribe() throws MQClientException {
        String nameServerAddress = getNameServerAddress();
        String consumerGroup = getConsumerGroup();
        MessageTopic topic = messageTopic();
        String tags = messageTags();
        tags = StringUtils.defaultIfBlank(tags,RocketMQConstants.MESSAGE_TAG_ALL);
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        // Specify name server addresses.
        consumer.setNamesrvAddr(nameServerAddress);
        // Subscribe one more more topics to consume.
        consumer.subscribe(topic.getTopic(), tags);
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> process(messageExtList));
        //Launch the consumer instance.
        consumer.start();
        log.info("开始消费消息，nameServerAddress:{}，consumerGroup:{}，topic:{}，tags:{}", nameServerAddress, consumerGroup, topic.getTopic(), tags);
    }

    /**
     * 消息处理监听
     * @param messageExtList 消息列表
     * @return 处理状态
     */
    private ConsumeConcurrentlyStatus process(List<MessageExt> messageExtList) {
        if (CollectionUtils.isEmpty(messageExtList)) {
            log.info("消息列表为空");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        if (messageExtList.size() > 1) {
            List<String> topics = messageExtList.stream().map(Message::getTopic).collect(Collectors.toList());
            log.error("消费者一次收到多个消息，topic:{}", topics);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        MessageExt messageExt = messageExtList.get(0);
        String topic = messageExt.getTopic();
        byte[] body = messageExt.getBody();
        String content = new String(body, StandardCharsets.UTF_8);
        boolean success = false;
        try {
            success = consume(content);
        } catch (Exception e) {
            log.error("消息消费失败，topic:{}，content:{}", topic, content);
        }
        if (!success) {
            return reConsume(messageExt);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 重复消费策略
     * @param messageExt 消息
     * @return 消息处理状态
     */
    private ConsumeConcurrentlyStatus reConsume(MessageExt messageExt) {
        String topic = messageExt.getTopic();
        int reConsumeTimes = messageExt.getReconsumeTimes();
        int reConsumeTimesMax = getReConsumeTimesMax();
        if (reConsumeTimes >= reConsumeTimesMax) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } else {
            log.info("消息第[{}]次重复消费失败，延后重复消费，topic:{}", reConsumeTimes, topic);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    /**
     * 消费者组
     */
    protected String getConsumerGroup() {
        return RocketMQConstants.CONSUMER_GROUP;
    }

    /**
     * NameServer地址
     */
    protected String getNameServerAddress() {
        return RocketMQConstants.NAME_SERVER_ADDRESS;
    }

    /**
     * 最大重复消费次数
     */
    protected int getReConsumeTimesMax() {
        return RocketMQConstants.RE_CONSUME_TIMES_MAX;
    }

    /**
     * 消息主题
     * @return 消息主题
     */
    protected abstract MessageTopic messageTopic();

    /**
     * 消息tags，*表示所有tag
     * @return 消息tags
     */
    protected String messageTags() {
        return RocketMQConstants.MESSAGE_TAG_ALL;
    }

    /**
     * 消息消费实现
     * @param content 消息内容
     * @return 是否消费成功
     */
    protected abstract boolean consume(String content);
}
